{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%run startup.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "$.getScript('./assets/js/ipython_notebook_toc.js')"
      ],
      "text/plain": [
       "<IPython.core.display.Javascript object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%javascript\n",
    "$.getScript('./assets/js/ipython_notebook_toc.js')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Decision Tree of Observable Operators\n",
    "\n",
    "## Part 5: Consolidating Streams\n",
    "\n",
    "> source: http://reactivex.io/documentation/operators.html#tree.  \n",
    "> (transcribed to RxPY 1.5.7, Py2.7 / 2016-12, Gunther Klessinger, [axiros](http://www.axiros.com))  \n",
    "\n",
    "**This tree can help you find the ReactiveX Observable operator you’re looking for.**  \n",
    "See [Part 1](./A Decision Tree of Observable Operators. Part I - Creation.ipynb) for Usage and Output Instructions.  \n",
    "\n",
    "We also require acquaintance with the [marble diagrams](./Marble Diagrams.ipynb) feature of RxPy.\n",
    "\n",
    "<h2 id=\"tocheading\">Table of Contents</h2>\n",
    "<div id=\"toc\"></div>\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want to reemit only certain items from an Observable"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by filtering out those that do not match some predicate **[filter/where](http://reactivex.io/documentation/operators/filter.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'reset_start_time' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-c3481d609448>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mreset_start_time\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mO\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;31m# alias: where\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0md\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0msubs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mO\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrange\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m5\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;32mlambda\u001b[0m \u001b[0mx\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mi\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mx\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0;36m2\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;36m0\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mNameError\u001b[0m: name 'reset_start_time' is not defined"
     ]
    }
   ],
   "source": [
    "reset_start_time(O.filter) # alias: where\n",
    "d = subs(O.range(0, 5).filter(lambda x, i: x % 2 == 0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by slicing **[slice](http://reactivex.io/documentation/operators/filter.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== slice_ ==========\n",
      "\n",
      "module rx.linq.observable.slice\n",
      "@extensionmethod(Observable, name=\"slice\")\n",
      "def slice_(self, start=None, stop=None, step=1):\n",
      "    Slices the given observable. It is basically a wrapper around the\n",
      "    operators skip(), skip_last(), take(), take_last() and filter().\n",
      "\n",
      "    This marble diagram helps you remember how slices works with streams.\n",
      "    Positive numbers is relative to the start of the events, while negative\n",
      "    numbers are relative to the end (on_completed) of the stream.\n",
      "\n",
      "    r---e---a---c---t---i---v---e---|\n",
      "    0   1   2   3   4   5   6   7   8\n",
      "   -8  -7  -6  -5  -4  -3  -2  -1\n",
      "\n",
      "    Example:\n",
      "    result = source.slice(1, 10)\n",
      "    result = source.slice(1, -2)\n",
      "    result = source.slice(1, -1, 2)\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param Observable self: Observable to slice\n",
      "    :param int start: Number of elements to skip of take last\n",
      "    :param int stop: Last element to take of skip last\n",
      "    :param int step: Takes every step element. Must be larger than zero\n",
      "\n",
      "    :returns: Returns a sliced observable sequence.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.2     M New subscription on stream 275034569\n",
      " 571.3   T13 [next]  568.0: i\n",
      " 681.0   T15 [next]  677.6: v\n",
      " 788.2   T17 [next]  784.8: e\n",
      " 898.9   T19 [cmpl]  895.6: fin\n",
      "\n",
      "1009.3     M New subscription on stream 276459417\n",
      "1245.7   T25 [next]  236.2: e\n",
      "1465.8   T30 [next]  456.3: c\n",
      "1686.8   T33 [next]  677.3: i\n",
      "1910.5   T37 [cmpl]  901.0: fin\n"
     ]
    }
   ],
   "source": [
    "reset_start_time(O.slice)\n",
    "s = marble_stream('r-e-a-c-t-i-v-e-|')\n",
    "d = subs(s.slice(5, 10))\n",
    "sleep(1)\n",
    "# start stop step:\n",
    "d = subs(s.slice(1, -1, 2))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that is, only the first item **[first](http://reactivex.io/documentation/operators/first.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== first ==========\n",
      "\n",
      "module rx.linq.observable.first\n",
      "@extensionmethod(Observable)\n",
      "def first(self, predicate=None):\n",
      "    Returns the first element of an observable sequence that satisfies\n",
      "    the condition in the predicate if present else the first item in the\n",
      "    sequence.\n",
      "\n",
      "    Example:\n",
      "    res = res = source.first()\n",
      "    res = res = source.first(lambda x: x > 3)\n",
      "\n",
      "    Keyword arguments:\n",
      "    predicate -- {Function} [Optional] A predicate function to evaluate for\n",
      "        elements in the source sequence.\n",
      "\n",
      "    Returns {Observable} Sequence containing the first element in the\n",
      "    observable sequence that satisfies the condition in the predicate if\n",
      "    provided, else the first item in the sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.2     M New subscription on stream 276538769\n",
      "   2.9     M [next]    0.6: 2\n",
      "   3.0     M [cmpl]    0.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.first)\n",
    "# match on index:\n",
    "d = subs(O.from_((1, 2 ,3)).first(lambda x, i: i==1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that is, only the first item*s* **[take, take_with_time](http://reactivex.io/documentation/operators/take.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== take ==========\n",
      "\n",
      "module rx.linq.observable.take\n",
      "@extensionmethod(Observable)\n",
      "def take(self, count, scheduler=None):\n",
      "    Returns a specified number of contiguous elements from the start of\n",
      "    an observable sequence, using the specified scheduler for the edge case\n",
      "    of take(0).\n",
      "\n",
      "    1 - source.take(5)\n",
      "    2 - source.take(0, rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    count -- The number of elements to return.\n",
      "    scheduler -- [Optional] Scheduler used to produce an OnCompleted\n",
      "        message in case count is set to 0.\n",
      "\n",
      "    Returns an observable sequence that contains the specified number of\n",
      "    elements from the start of the input sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.0     M New subscription on stream 276541045\n",
      "   3.6     M [next]    0.5: 1\n",
      "   3.8     M [next]    0.7: 2\n",
      "   4.0     M [cmpl]    0.9: fin\n",
      "\n",
      "\n",
      "========== take_with_time ==========\n",
      "\n",
      "module rx.linq.observable.takewithtime\n",
      "@extensionmethod(Observable)\n",
      "def take_with_time(self, duration, scheduler=None):\n",
      "    Takes elements for the specified duration from the start of the\n",
      "    observable source sequence, using the specified scheduler to run timers.\n",
      "\n",
      "    Example:\n",
      "    res = source.take_with_time(5000,  [optional scheduler])\n",
      "\n",
      "    Description:\n",
      "    This operator accumulates a queue with a length enough to store elements\n",
      "    received during the initial duration window. As more elements are\n",
      "    received, elements older than the specified duration are taken from the\n",
      "    queue and produced on the result sequence. This causes elements to be\n",
      "    delayed with duration.\n",
      "\n",
      "    Keyword arguments:\n",
      "    duration -- {Number} Duration for taking elements from the start of the\n",
      "        sequence.\n",
      "    scheduler -- {Scheduler} Scheduler to run the timer on. If not\n",
      "        specified, defaults to rx.Scheduler.timeout.\n",
      "\n",
      "    Returns {Observable} An observable sequence with the elements taken\n",
      "    during the specified duration from the start of the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.4     M New subscription on stream 276540693\n",
      "  13.6  T150 [next]   12.1: 1\n",
      " 123.5  T151 [next]  122.0: 2\n",
      " 204.9  T149 [cmpl]  203.4: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.take)\n",
    "d = subs(O.from_((1, 2, 3, 4)).take(2))\n",
    "rst(O.take_with_time)\n",
    "d = subs(marble_stream('1-2-3-4|').take_with_time(200))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that is, only the last item **[last, last_or_default, take_last](http://reactivex.io/documentation/operators/last.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== last ==========\n",
      "\n",
      "module rx.linq.observable.last\n",
      "@extensionmethod(Observable)\n",
      "def last(self, predicate=None):\n",
      "    Returns the last element of an observable sequence that satisfies the\n",
      "    condition in the predicate if specified, else the last element.\n",
      "\n",
      "    Example:\n",
      "    res = source.last()\n",
      "    res = source.last(lambda x: x > 3)\n",
      "\n",
      "    Keyword arguments:\n",
      "    predicate -- {Function} [Optional] A predicate function to evaluate for\n",
      "        elements in the source sequence.\n",
      "\n",
      "    Returns {Observable} Sequence containing the last element in the\n",
      "    observable sequence that satisfies the condition in the predicate.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.5     M New subscription on stream 276538821\n",
      "   3.2     M [next]    0.6: 2\n",
      "   3.4     M [cmpl]    0.8: fin\n",
      "\n",
      "\n",
      "========== last_or_default ==========\n",
      "\n",
      "module rx.linq.observable.lastordefault\n",
      "@extensionmethod(Observable)\n",
      "def last_or_default(self, predicate=None, default_value=None):\n",
      "    Return last or default element.\n",
      "\n",
      "    Returns the last element of an observable sequence that satisfies\n",
      "    the condition in the predicate, or a default value if no such\n",
      "    element exists.\n",
      "\n",
      "    Examples:\n",
      "    res = source.last_or_default()\n",
      "    res = source.last_or_default(lambda x: x > 3)\n",
      "    res = source.last_or_default(lambda x: x > 3, 0)\n",
      "    res = source.last_or_default(None, 0)\n",
      "\n",
      "    predicate -- {Function} [Optional] A predicate function to evaluate\n",
      "        for elements in the source sequence.\n",
      "    default_value -- [Optional] The default value if no such element\n",
      "        exists. If not specified, defaults to None.\n",
      "\n",
      "    Returns {Observable} Sequence containing the last element in the\n",
      "    observable sequence that satisfies the condition in the predicate,\n",
      "    or a default value if no such element exists.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.4     M New subscription on stream 276540749\n",
      "   2.0     M [next]    0.5: None\n",
      "   2.1     M [cmpl]    0.6: fin\n",
      "\n",
      "   2.7     M New subscription on stream 276540757\n",
      "   3.3     M [next]    0.5: 42\n",
      "   3.4     M [cmpl]    0.6: fin\n",
      "\n",
      "\n",
      "========== take_last ==========\n",
      "\n",
      "module rx.linq.observable.takelast\n",
      "@extensionmethod(Observable)\n",
      "def take_last(self, count):\n",
      "    Returns a specified number of contiguous elements from the end of an\n",
      "    observable sequence.\n",
      "\n",
      "    Example:\n",
      "    res = source.take_last(5)\n",
      "\n",
      "    Description:\n",
      "    This operator accumulates a buffer with a length enough to store\n",
      "    elements count elements. Upon completion of the source sequence, this\n",
      "    buffer is drained on the result sequence. This causes the elements to be\n",
      "    delayed.\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param int count: Number of elements to take from the end of the source\n",
      "        sequence.\n",
      "\n",
      "    :returns: An observable sequence containing the specified number of elements \n",
      "        from the end of the source sequence.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.7     M New subscription on stream 276541765\n",
      "   2.4     M [next]    0.6: 3\n",
      "   2.5     M [next]    0.7: 4\n",
      "   2.6     M [cmpl]    0.8: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.last,  title=True)\n",
    "d = subs(O.from_((1, 2, 3)).last(lambda x: x < 3))\n",
    "rst(O.last_or_default,  title=True)\n",
    "d = subs(O.from_((1, 2, 3)).last_or_default(lambda x: x > 3))\n",
    "d = subs(O.from_((1, 2, 3)).last_or_default(lambda x: x > 3, '42'))\n",
    "\n",
    "rst(O.take_last, title=True)\n",
    "d = subs(O.from_((1, 2, 3, 4)).take_last(2))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that is, only item n **[element_at, element_at_or_default](http://reactivex.io/documentation/operators/elementat.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== element_at ==========\n",
      "\n",
      "module rx.linq.observable.elementat\n",
      "@extensionmethod(Observable)\n",
      "def element_at(self, index):\n",
      "    Returns the element at a specified index in a sequence.\n",
      "\n",
      "    Example:\n",
      "    res = source.element_at(5)\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param int index: The zero-based index of the element to retrieve.\n",
      "\n",
      "    :returns: An observable  sequence that produces the element at the\n",
      "    specified position in the source sequence.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.2     M New subscription on stream 278634469\n",
      "   2.8     M [next]    0.5: 3\n",
      "   2.9     M [cmpl]    0.6: fin\n",
      "\n",
      "\n",
      "========== element_at_or_default ==========\n",
      "\n",
      "module rx.linq.observable.elementatordefault\n",
      "@extensionmethod(Observable)\n",
      "def element_at_or_default(self, index, default_value=None):\n",
      "    Returns the element at a specified index in a sequence or a default\n",
      "    value if the index is out of range.\n",
      "\n",
      "    Example:\n",
      "    res = source.element_at_or_default(5)\n",
      "    res = source.element_at_or_default(5, 0)\n",
      "\n",
      "    Keyword arguments:\n",
      "    index -- {Number} The zero-based index of the element to retrieve.\n",
      "    default_value -- [Optional] The default value if the index is outside\n",
      "        the bounds of the source sequence.\n",
      "\n",
      "    Returns an observable {Observable} sequence that produces the element at\n",
      "        the specified position in the source sequence, or a default value if\n",
      "        the index is outside the bounds of the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.3     M New subscription on stream 276533629\n",
      "   2.1     M [next]    0.6: 42\n",
      "   2.2     M [cmpl]    0.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.element_at)\n",
    "d = subs(O.from_((1, 2, 3, 4)).element_at(2))\n",
    "rst(O.element_at_or_default)\n",
    "d = subs(O.from_((1, 2, 3, 4)).element_at_or_default(6, '42'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that is, only those items after the first items"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... that is, after the first n items **[skip, skip_with_time](http://reactivex.io/documentation/operators/skip.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== skip ==========\n",
      "\n",
      "module rx.linq.observable.skip\n",
      "@extensionmethod(Observable)\n",
      "def skip(self, count):\n",
      "    Bypasses a specified number of elements in an observable sequence\n",
      "    and then returns the remaining elements.\n",
      "\n",
      "    Keyword arguments:\n",
      "    count -- The number of elements to skip before returning the remaining\n",
      "        elements.\n",
      "\n",
      "    Returns an observable sequence that contains the elements that occur\n",
      "    after the specified index in the input sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.5     M New subscription on stream 276082845\n",
      "   3.1     M [next]    0.5: 2\n",
      "   3.3     M [next]    0.7: 3\n",
      "   3.7     M [next]    1.1: 4\n",
      "   3.9     M [cmpl]    1.3: fin\n",
      "\n",
      "\n",
      "========== skip_with_time ==========\n",
      "\n",
      "module rx.linq.observable.skipwithtime\n",
      "@extensionmethod(Observable)\n",
      "def skip_with_time(self, duration, scheduler=None):\n",
      "    Skips elements for the specified duration from the start of the\n",
      "    observable source sequence, using the specified scheduler to run timers.\n",
      "\n",
      "    Example:\n",
      "    1 - res = source.skip_with_time(5000, [optional scheduler])\n",
      "\n",
      "    Description:\n",
      "    Specifying a zero value for duration doesn't guarantee no elements will\n",
      "    be dropped from the start of the source sequence. This is a side-effect\n",
      "    of the asynchrony introduced by the scheduler, where the action that\n",
      "    causes callbacks from the source sequence to be forwarded may not\n",
      "    execute immediately, despite the zero due time.\n",
      "\n",
      "    Errors produced by the source sequence are always forwarded to the\n",
      "    result sequence, even if the error occurs before the duration.\n",
      "\n",
      "    Keyword arguments:\n",
      "    duration -- {Number} Duration for skipping elements from the start of\n",
      "        the sequence.\n",
      "    scheduler -- {Scheduler} Scheduler to run the timer on. If not\n",
      "        specified, defaults to Rx.Scheduler.timeout.\n",
      "\n",
      "    Returns n observable {Observable} sequence with the elements skipped\n",
      "    during the specified duration from the start of the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.7     M New subscription on stream 276541889\n",
      " 236.1  T165 [next]  234.4: 3\n",
      " 347.9  T167 [next]  346.1: 4\n",
      " 455.7  T168 [next]  453.9: 5\n",
      " 565.7  T170 [next]  563.9: 6\n",
      " 566.5  T172 [cmpl]  564.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.skip, title=True)\n",
    "d = subs(O.range(0, 5).skip(2))\n",
    "rst(O.skip_with_time, title=True)\n",
    "d = subs(marble_stream('1-2-3-4-5-6').skip_with_time(200))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... that is, until one of those items matches a predicate **[skip_while](http://reactivex.io/documentation/operators/skipwhile.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== skip_while ==========\n",
      "\n",
      "module rx.linq.observable.skipwhile\n",
      "@extensionmethod(Observable)\n",
      "def skip_while(self, predicate):\n",
      "    Bypasses elements in an observable sequence as long as a specified\n",
      "    condition is true and then returns the remaining elements. The\n",
      "    element's index is used in the logic of the predicate function.\n",
      "\n",
      "    1 - source.skip_while(lambda value: value < 10)\n",
      "    2 - source.skip_while(lambda value, index: value < 10 or index < 10)\n",
      "\n",
      "    predicate -- A function to test each element for a condition; the\n",
      "        second parameter of the function represents the index of the\n",
      "        source element.\n",
      "\n",
      "    Returns an observable sequence that contains the elements from the\n",
      "    input sequence starting at the first element in the linear series that\n",
      "    does not pass the test specified by predicate.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.7     M New subscription on stream 275707413\n",
      "   2.2     M [next]    0.5: 3\n",
      "   2.5     M [next]    0.8: 4\n",
      "   2.8     M [next]    1.1: 5\n",
      "   3.1     M [next]    1.4: 6\n",
      "   3.4     M [cmpl]    1.6: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.skip_while)\n",
    "# skipping only AS LONG AS the function is true. If already false at the beginning -> all flushed:\n",
    "d = subs(O.from_((1, 2, 3, 4, 5, 6)).skip_while(lambda x: x in (1, 2)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... that is, after a second Observable emits an item **[skip_until, skip_until_with_time](http://reactivex.io/documentation/operators/skipuntil.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== skip_until ==========\n",
      "\n",
      "module rx.linq.observable.skipuntil\n",
      "@extensionmethod(Observable)\n",
      "def skip_until(self, other):\n",
      "    Returns the values from the source observable sequence only after\n",
      "    the other observable sequence produces a value.\n",
      "\n",
      "    other -- The observable sequence that triggers propagation of elements\n",
      "        of the source sequence.\n",
      "\n",
      "    Returns an observable sequence containing the elements of the source\n",
      "    sequence starting from the point the other sequence triggered\n",
      "    propagation.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.8     M New subscription on stream 276474365\n",
      " 235.7    T8 [next]  232.9: 3\n",
      " 346.9   T10 [next]  344.1: 4\n",
      " 456.9   T12 [next]  454.1: 5\n",
      " 466.9   T13 [cmpl]  464.1: fin\n",
      "\n",
      "\n",
      "========== skip_until_with_time ==========\n",
      "\n",
      "module rx.linq.observable.skipuntilwithtime\n",
      "@extensionmethod(Observable)\n",
      "def skip_until_with_time(self, start_time, scheduler=None):\n",
      "    Skips elements from the observable source sequence until the\n",
      "    specified start time, using the specified scheduler to run timers.\n",
      "    Errors produced by the source sequence are always forwarded to the\n",
      "    result sequence, even if the error occurs before the start time.\n",
      "\n",
      "    Examples:\n",
      "    res = source.skip_until_with_time(new Date(), [optional scheduler]);\n",
      "    res = source.skip_until_with_time(5000, [optional scheduler]);\n",
      "\n",
      "    Keyword arguments:\n",
      "    start_time -- Time to start taking elements from the source sequence. If\n",
      "        this value is less than or equal to Date(), no elements will be\n",
      "        skipped.\n",
      "    scheduler -- Scheduler to run the timer on. If not specified, defaults\n",
      "        to rx.Scheduler.timeout.\n",
      "\n",
      "    Returns {Observable} An observable sequence with the elements skipped\n",
      "    until the specified start time.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.2     M New subscription on stream 276064481\n",
      " 348.3   T26 [next]  344.9: 4\n",
      " 461.7   T28 [next]  458.4: 5\n",
      " 468.8   T29 [cmpl]  465.5: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.skip_until)\n",
    "s1 = marble_stream('1-2-3-4-5|')\n",
    "s2 = marble_stream('--2------|')\n",
    "d = subs(s1.skip_until(s2))\n",
    "sleep(0.5)\n",
    "rst(O.skip_until_with_time)\n",
    "d = subs(s1.skip_until_with_time(300))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that is, those items except the last items"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... that is, except the last n items **[skip_last, skip_last_with_time](http://reactivex.io/documentation/operators/skiplast.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== skip_last ==========\n",
      "\n",
      "module rx.linq.observable.skiplast\n",
      "@extensionmethod(Observable)\n",
      "def skip_last(self, count):\n",
      "    Bypasses a specified number of elements at the end of an observable\n",
      "    sequence.\n",
      "\n",
      "    Description:\n",
      "    This operator accumulates a queue with a length enough to store the\n",
      "    first `count` elements. As more elements are received, elements are\n",
      "    taken from the front of the queue and produced on the result sequence.\n",
      "    This causes elements to be delayed.\n",
      "\n",
      "    Keyword arguments\n",
      "    count -- Number of elements to bypass at the end of the source sequence.\n",
      "\n",
      "    Returns an observable {Observable} sequence containing the source\n",
      "    sequence elements except for the bypassed ones at the end.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.6     M New subscription on stream 276525625\n",
      " 234.8   T31 [next]  232.2: 1\n",
      " 345.4   T33 [next]  342.8: 2\n",
      " 455.7   T36 [next]  453.0: 3\n",
      " 469.9   T37 [cmpl]  467.3: fin\n",
      "\n",
      "\n",
      "========== skip_last_with_time ==========\n",
      "\n",
      "module rx.linq.observable.skiplastwithtime\n",
      "@extensionmethod(Observable)\n",
      "def skip_last_with_time(self, duration, scheduler=None):\n",
      "    Skips elements for the specified duration from the end of the\n",
      "    observable source sequence, using the specified scheduler to run timers.\n",
      "\n",
      "    1 - res = source.skip_last_with_time(5000)\n",
      "    2 - res = source.skip_last_with_time(5000, scheduler)\n",
      "\n",
      "    Description:\n",
      "    This operator accumulates a queue with a length enough to store elements\n",
      "    received during the initial duration window. As more elements are\n",
      "    received, elements older than the specified duration are taken from the\n",
      "    queue and produced on the result sequence. This causes elements to be\n",
      "    delayed with duration.\n",
      "\n",
      "    Keyword arguments:\n",
      "    duration -- {Number} Duration for skipping elements from the end of the\n",
      "        sequence.\n",
      "    scheduler -- {Scheduler} [Optional]  Scheduler to run the timer on. If\n",
      "        not specified, defaults to Rx.Scheduler.timeout\n",
      "    Returns an observable {Observable} sequence with the elements skipped\n",
      "    during the specified duration from the end of the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.9     M New subscription on stream 276128193\n",
      " 348.3   T46 [next]  344.2: 1\n",
      " 459.2   T47 [next]  455.2: 2\n",
      " 471.7   T50 [cmpl]  467.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.skip_last)\n",
    "s1 = marble_stream('1-2-3-4-5|')\n",
    "s2 = marble_stream('--2------|')\n",
    "d = subs(s1.skip_last(2))\n",
    "sleep(0.5)\n",
    "rst(O.skip_last_with_time)\n",
    "d = subs(s1.skip_last_with_time(300))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... that is, until one of those items matches a predicate **[take_while](http://reactivex.io/documentation/operators/takewhile.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== take_while ==========\n",
      "\n",
      "module rx.linq.observable.takewhile\n",
      "@extensionmethod(Observable)\n",
      "def take_while(self, predicate):\n",
      "    Returns elements from an observable sequence as long as a specified\n",
      "    condition is true. The element's index is used in the logic of the\n",
      "    predicate function.\n",
      "\n",
      "    1 - source.take_while(lambda value: value < 10)\n",
      "    2 - source.take_while(lambda value, index: value < 10 or index < 10)\n",
      "\n",
      "    Keyword arguments:\n",
      "    predicate -- A function to test each element for a condition; the\n",
      "        second parameter of the function represents the index of the source\n",
      "        element.\n",
      "\n",
      "    Returns an observable sequence that contains the elements from the\n",
      "    input sequence that occur before the element at which the test no\n",
      "    longer passes.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.6     M New subscription on stream 278900289\n",
      "   2.9     M [next]    0.3: 1\n",
      "   3.5     M [next]    0.8: 2\n",
      "   4.0     M [cmpl]    1.4: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.take_while)\n",
    "d = subs(O.from_((1, 2, 3)).take_while(lambda x: x<3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ...that is, except items emitted during a period of time before the source completes **[skip_last, skip_last_with_time](http://reactivex.io/documentation/operators/skiplast.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# (see above)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ...that is, except items emitted after a second Observable emits an item **[take_until, take_until_with_time](http://reactivex.io/documentation/operators/takeuntil.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== take_until ==========\n",
      "\n",
      "module rx.linq.observable.takeuntil\n",
      "@extensionmethod(Observable)\n",
      "def take_until(self, other):\n",
      "    Returns the values from the source observable sequence until the\n",
      "    other observable sequence produces a value.\n",
      "\n",
      "    Keyword arguments:\n",
      "    other -- Observable sequence that terminates propagation of elements of\n",
      "        the source sequence.\n",
      "\n",
      "    Returns an observable sequence containing the elements of the source\n",
      "    sequence up to the point the other sequence interrupted further\n",
      "    propagation.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.1     M New subscription on stream 278898461\n",
      "  13.8  T131 [next]   11.6: 1\n",
      " 123.9  T133 [next]  121.7: 2\n",
      " 216.0  T143 [cmpl]  213.8: fin\n",
      "\n",
      "\n",
      "========== take_until_with_time ==========\n",
      "\n",
      "module rx.linq.observable.takeuntilwithtime\n",
      "@extensionmethod(Observable)\n",
      "def take_until_with_time(self, end_time, scheduler=None):\n",
      "    Takes elements for the specified duration until the specified end\n",
      "    time, using the specified scheduler to run timers.\n",
      "\n",
      "    Examples:\n",
      "    1 - res = source.take_until_with_time(dt, [optional scheduler])\n",
      "    2 - res = source.take_until_with_time(5000, [optional scheduler])\n",
      "\n",
      "    Keyword Arguments:\n",
      "    end_time -- {Number | Date} Time to stop taking elements from the source\n",
      "        sequence. If this value is less than or equal to Date(), the\n",
      "        result stream will complete immediately.\n",
      "    scheduler -- {Scheduler} Scheduler to run the timer on.\n",
      "\n",
      "    Returns an observable {Observable} sequence with the elements taken\n",
      "    until the specified end time.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.1     M New subscription on stream 278906801\n",
      "  14.7  T148 [next]   12.5: 1\n",
      " 128.0  T149 [next]  125.8: 2\n",
      " 235.5  T152 [next]  233.4: 3\n",
      " 307.8  T147 [cmpl]  305.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.take_until)\n",
    "s1 = marble_stream('1-2-3-4-5|')\n",
    "s2 = marble_stream('--2------|')\n",
    "d = subs(s1.take_until(s2))\n",
    "sleep(0.5)\n",
    "rst(O.take_until_with_time)\n",
    "d = subs(s1.take_until_with_time(300))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by sampling the Observable periodically **[sample](http://reactivex.io/documentation/operators/sample.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== sample ==========\n",
      "\n",
      "module rx.linq.observable.sample\n",
      "@extensionmethod(Observable, alias=\"throttle_last\")\n",
      "def sample(self, interval=None, sampler=None, scheduler=None):\n",
      "    Samples the observable sequence at each interval.\n",
      "\n",
      "    1 - res = source.sample(sample_observable) # Sampler tick sequence\n",
      "    2 - res = source.sample(5000) # 5 seconds\n",
      "    2 - res = source.sample(5000, rx.scheduler.timeout) # 5 seconds\n",
      "\n",
      "    Keyword arguments:\n",
      "    source -- Source sequence to sample.\n",
      "    interval -- Interval at which to sample (specified as an integer\n",
      "        denoting milliseconds).\n",
      "    scheduler -- [Optional] Scheduler to run the sampling timer on. If not\n",
      "        specified, the timeout scheduler is used.\n",
      "\n",
      "    Returns sampled observable sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.9     M New subscription on stream 288677153\n",
      " 314.2  T847 [next]  312.2: 3\n",
      " 619.4  T849 [next]  617.4: 6\n",
      " 921.6  T850 [next]  919.6: 9\n",
      "1226.2  T851 [next] 1224.1: 2\n",
      "1530.3  T852 [next] 1528.2: 5\n",
      "1836.2  T853 [next] 1834.1: E\n",
      "1836.9  T853 [cmpl] 1834.9: fin\n",
      "\n",
      "2018.3     M New subscription on stream 279189229\n",
      "2338.6  T889 [next]  320.3: 3\n",
      "2651.7  T890 [next]  633.4: 6\n",
      "3661.6  T893 [next] 1643.3: 6\n",
      "4872.2  T894 [next] 2853.8: E\n",
      "4872.5  T894 [cmpl] 2854.2: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.sample)\n",
    "xs =     marble_stream('1-2-3-4-5-6-7-8-9-1-2-3-4-5-6-E|')\n",
    "sampler =marble_stream('---1---1----------1------------|')\n",
    "d = subs(xs.sample(300))\n",
    "sleep(2)\n",
    "d = subs(xs.sample(sampler=sampler))\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by only emitting items that are not followed by other items within some duration **[debounce](http://reactivex.io/documentation/operators/debounce.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== debounce ==========\n",
      "\n",
      "module rx.linq.observable.debounce\n",
      "@extensionmethod(Observable, alias=\"throttle_with_timeout\")\n",
      "def debounce(self, duetime, scheduler=None):\n",
      "    Ignores values from an observable sequence which are followed by\n",
      "    another value before duetime.\n",
      "\n",
      "    Example:\n",
      "    1 - res = source.debounce(5000) # 5 seconds\n",
      "    2 - res = source.debounce(5000, scheduler)\n",
      "\n",
      "    Keyword arguments:\n",
      "    duetime -- {Number} Duration of the throttle period for each value\n",
      "        (specified as an integer denoting milliseconds).\n",
      "    scheduler -- {Scheduler} [Optional]  Scheduler to run the throttle\n",
      "        timers on. If not specified, the timeout scheduler is used.\n",
      "\n",
      "    Returns {Observable} The debounced sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "flushing a value every >= 300ms\n",
      "\n",
      "   3.4     M New subscription on stream 281048569\n",
      "1062.2  T977 [next] 1058.8: 6\n",
      "1368.6  T978 [next] 1365.2: 7\n",
      "1685.5  T979 [next] 1682.1: 8\n",
      "2092.7  T980 [next] 2089.4: 9\n",
      "2203.3  T970 [cmpl] 2199.9: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.debounce)\n",
    "s = marble_stream('-12-3-4--5--6---7---8----9----a')\n",
    "print('flushing a value every >= 300ms')\n",
    "d = subs(s.debounce(300))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by suppressing items that are duplicates of already-emitted items **[distinct](http://reactivex.io/documentation/operators/distinct.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== distinct ==========\n",
      "\n",
      "module rx.linq.observable.distinct\n",
      "@extensionmethod(Observable)\n",
      "def distinct(self, key_selector=None, comparer=None):\n",
      "    Returns an observable sequence that contains only distinct elements\n",
      "    according to the key_selector and the comparer. Usage of this operator\n",
      "    should be considered carefully due to the maintenance of an internal\n",
      "    lookup structure which can grow large.\n",
      "\n",
      "    Example:\n",
      "    res = obs = xs.distinct()\n",
      "    obs = xs.distinct(lambda x: x.id)\n",
      "    obs = xs.distinct(lambda x: x.id, lambda a,b: a == b)\n",
      "\n",
      "    Keyword arguments:\n",
      "    key_selector -- {Function} [Optional]  A function to compute the\n",
      "        comparison key for each element.\n",
      "    comparer -- {Function} [Optional]  Used to compare items in the\n",
      "        collection.\n",
      "\n",
      "    Returns an observable {Observable} sequence only containing the distinct\n",
      "    elements, based on a computed key value, from the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.6     M New subscription on stream 286822853\n",
      "   3.0     M [next]    0.3: 1\n",
      "   3.2     M [next]    0.5: 2\n",
      "   3.5     M [next]    0.9: 3\n",
      "   3.8     M [cmpl]    1.1: fin\n",
      "\n",
      "   4.3     M New subscription on stream 286822721\n",
      "   4.6     M [next]    0.2: 1\n",
      "   4.9     M [next]    0.5: 2\n",
      "   5.4     M [cmpl]    1.0: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.distinct)\n",
    "s = O.from_((1, 2, 1, 1, 3))\n",
    "d = subs(s.distinct(lambda x: x*2))\n",
    "d = subs(s.distinct(lambda x: x, lambda a, b: a==2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... if they immediately follow the item they are duplicates of **[distinct_until_changed](http://reactivex.io/documentation/operators/distinct.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== distinct_until_changed ==========\n",
      "\n",
      "module rx.linq.observable.distinctuntilchanged\n",
      "@extensionmethod(Observable)\n",
      "def distinct_until_changed(self, key_selector=None, comparer=None):\n",
      "    Returns an observable sequence that contains only distinct\n",
      "    contiguous elements according to the key_selector and the comparer.\n",
      "\n",
      "    1 - obs = observable.distinct_until_changed();\n",
      "    2 - obs = observable.distinct_until_changed(lambda x: x.id)\n",
      "    3 - obs = observable.distinct_until_changed(lambda x: x.id,\n",
      "                                                lambda x, y: x == y)\n",
      "\n",
      "    key_selector -- [Optional] A function to compute the comparison key for\n",
      "        each element. If not provided, it projects the value.\n",
      "    comparer -- [Optional] Equality comparer for computed key values. If\n",
      "        not provided, defaults to an equality comparer function.\n",
      "\n",
      "    Return An observable sequence only containing the distinct contiguous\n",
      "    elements, based on a computed key value, from the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.2     M New subscription on stream 281042609\n",
      "   4.1     M [next]    0.8: 1\n",
      "   5.0     M [next]    1.7: 2\n",
      "   5.4     M [next]    2.1: 1\n",
      "   5.9     M [next]    2.6: 3\n",
      "   6.2     M [cmpl]    2.9: fin\n",
      "\n",
      "   7.0     M New subscription on stream 286831301\n",
      "   7.5     M [next]    0.3: 1\n",
      "   7.8     M [next]    0.7: 2\n",
      "   8.2     M [cmpl]    1.0: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.distinct_until_changed)\n",
    "s = O.from_((1, 2, 1, 1, 3))\n",
    "d = subs(s.distinct_until_changed(lambda x: x*2))\n",
    "d = subs(s.distinct_until_changed(lambda x: x, lambda a, b: a==2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... by delaying my subscription to it for some time after it begins emitting items **[delay_subscription](http://reactivex.io/documentation/operators/delay.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== delay ==========\n",
      "\n",
      "module rx.linq.observable.delay\n",
      "@extensionmethod(Observable)\n",
      "def delay(self, duetime, scheduler=None):\n",
      "    Time shifts the observable sequence by duetime. The relative time\n",
      "    intervals between the values are preserved.\n",
      "\n",
      "    1 - res = rx.Observable.delay(datetime())\n",
      "    2 - res = rx.Observable.delay(datetime(), Scheduler.timeout)\n",
      "\n",
      "    3 - res = rx.Observable.delay(5000)\n",
      "    4 - res = rx.Observable.delay(5000, Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param datetime|int duetime: Absolute (specified as a datetime object) or\n",
      "        relative time (specified as an integer denoting milliseconds) by which\n",
      "        to shift the observable sequence.\n",
      "    :param Scheduler scheduler: [Optional] Scheduler to run the delay timers on.\n",
      "        If not specified, the timeout scheduler is used.\n",
      "\n",
      "    :returns: Time-shifted sequence.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "\n",
      "========== note the absolute time of emissions: ==========\n",
      "\n",
      "\n",
      "   2.1     M New subscription on stream 276542741\n",
      "1005.6   T11 [next] 1003.4: 0\n",
      "1005.9   T11 [next] 1003.7: 1\n",
      "1006.1   T11 [next] 1003.9: 2\n",
      "1006.2   T11 [next] 1003.9: 3\n",
      "1006.2   T11 [next] 1004.0: 4\n",
      "1006.3   T11 [next] 1004.0: 5\n",
      "1006.3   T11 [next] 1004.1: 6\n",
      "1006.4   T11 [next] 1004.1: 7\n",
      "1006.4   T11 [next] 1004.1: 8\n",
      "1006.4   T11 [next] 1004.2: 9\n",
      "1006.5   T11 [cmpl] 1004.2: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.delay)\n",
    "header(\"note the absolute time of emissions:\")\n",
    "d = subs(O.range(0, 10).delay(1000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# I want to reemit items from an Observable only on condition that it was the first of a collection of Observables to emit an item **[amb](http://reactivex.io/documentation/operators/amb.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== amb ==========\n",
      "\n",
      "module rx.linq.observable.amb\n",
      "@extensionclassmethod(Observable)\n",
      "def amb(cls, *args):\n",
      "    Propagates the observable sequence that reacts first.\n",
      "\n",
      "    E.g. winner = rx.Observable.amb(xs, ys, zs)\n",
      "\n",
      "    Returns an observable sequence that surfaces any of the given sequences,\n",
      "    whichever reacted first.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.4     M New subscription on stream 276511041\n",
      "   6.4     M [next]    2.8: 10\n",
      "   6.7     M [next]    3.0: 11\n",
      "   7.0     M [next]    3.4: 12\n",
      "   7.5     M [next]    3.9: 13\n",
      "   7.9     M [next]    4.2: 14\n",
      "   8.2     M [cmpl]    4.6: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.amb)\n",
    "s1 = O.range(0, 5).delay(100)\n",
    "s2 = O.range(10, 5)\n",
    "\n",
    "d = subs(O.amb(s1, s2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
